
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/redirect_error.hpp>

#include "flv_media_sink.hpp"
#include "flv_media_source.hpp"

#include "server/rtmp/rtmp_protocol/rtmp_define.hpp"
#include "server/rtmp/rtmp_message/data_message/rtmp_metadata_message.hpp"

#include "base/thread/thread_worker.hpp"
#include "base/sequence_pkt_buf.hpp"
#include "spdlog/spdlog.h"

using namespace mms;

FlvMediaSink::FlvMediaSink(ThreadWorker* worker) : MediaSink(worker) {
}

bool FlvMediaSink::init() {
    return true;
}

FlvMediaSink::~FlvMediaSink() {
    spdlog::info("destroy FlvMediaSink");
}

bool FlvMediaSink::on_audio_packet(std::shared_ptr<FlvTag> audio_pkt) {
    auto self(this->shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self, audio_pkt]()->boost::asio::awaitable<void> {
        std::vector<std::shared_ptr<FlvTag>> pkts{ audio_pkt };
        if (!co_await cb_(pkts)) {
            co_return;
        }
        co_return;
        }, boost::asio::detached);
    return true;
}

bool FlvMediaSink::on_video_packet(std::shared_ptr<FlvTag> video_pkt) {
    auto self(this->shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self, video_pkt]()->boost::asio::awaitable<void> {
        std::vector<std::shared_ptr<FlvTag>> pkts{ video_pkt };
        if (!co_await cb_(pkts)) {
            co_return;
        }
        co_return;
        }, boost::asio::detached);
    return true;
}

boost::asio::awaitable<bool> FlvMediaSink::send_flv_tag(std::shared_ptr<FlvTag> pkt) {
    ((void)pkt);
    co_return true;
}

bool FlvMediaSink::on_metadata(std::shared_ptr<FlvTag> metadata_pkt) {
    auto self(this->shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self, metadata_pkt]()->boost::asio::awaitable<void> {
        std::vector<std::shared_ptr<FlvTag>> pkts{ metadata_pkt };
        if (!co_await cb_(pkts)) {
            co_return;
        }
        co_return;
        }, boost::asio::detached);
    return true;
}

void FlvMediaSink::wakeup() {
    if (working_) {
        return;
    }
    working_ = true;
    auto self(this->shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self]()->boost::asio::awaitable<void> {
        if (!source_->is_ready()) {
            working_ = false;
            spdlog::debug("source is not ready");
            co_return;
        }

        std::vector<std::shared_ptr<FlvTag>> pkts;
        auto flv_source = std::static_pointer_cast<FlvMediaSource>(source_);
        do {
            pkts = flv_source->get_pkts(last_send_pkt_index_, 10);
            if (!co_await cb_(pkts)) {
                break;
            }
        } while (pkts.size() > 0);
        working_ = false;
        co_return;
        }, boost::asio::detached);
}

void FlvMediaSink::on_flv_tag(const std::function<boost::asio::awaitable<bool>(std::vector<std::shared_ptr<FlvTag>>& msgs)>& cb) {
    cb_ = cb;
}

void FlvMediaSink::close() {
    cb_ = [](const std::vector<std::shared_ptr<FlvTag>>& msgs)->boost::asio::awaitable<bool> {
        ((void)msgs);
        co_return true;
        };
    auto self(shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self]()->boost::asio::awaitable<void> {
        spdlog::info(" source_->remove_media_sink(self);");
        source_->remove_media_sink(self);
        co_return;
        }, boost::asio::detached);
    MediaSink::close();
}